package com.zxmr.action;

import java.io.IOException;
import java.util.concurrent.ExecutorService;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.ShutdownSignalException;

public abstract class QueueReceiver implements Runnable{

	boolean mainloopRunning = true;
	
	/*
	 *  thread pool that used to execute detail task
	*/
	protected ExecutorService es = null;
	
	protected String queue = null;
	protected Channel channel = null;
	protected QueueingConsumer consumer = null;
	
	public QueueReceiver(ExecutorService es,String queue)
	{
		this.es = es;
		this.queue = queue;
	}

    protected QueueingConsumer setupConsumer()throws IOException
    {
    	QueueingConsumer consumer = new QueueingConsumer(channel);
    	channel.basicConsume(queue, true,consumer);
    	return consumer;
    }
	@Override
	public void run() {

        try {
        	channel = setupChannel();
			consumer = setupConsumer();
        	
            while (mainloopRunning) {
                QueueingConsumer.Delivery request;
                try {
                    request = consumer.nextDelivery();
                } catch (InterruptedException ie) {
                    continue;
                }
                processRequestPerThread(request);
            }
        } catch (ShutdownSignalException sse) {
            sse.printStackTrace();
        }catch(IOException ex)
        {
        	ex.printStackTrace();
        }catch(Exception e)
        {
        	e.printStackTrace();
        }
        
	}
	private void processRequestPerThread(final QueueingConsumer.Delivery request)throws IOException
	{
		es.execute(new Runnable()
		{

			@Override
			public void run() {

				try {
					processRequest(request);
				} catch (IOException e) {
					e.printStackTrace();
				}catch(Exception ee)
				{
					ee.printStackTrace();
				}
			}
			
		});
	}
	public abstract void processRequest(QueueingConsumer.Delivery request)throws IOException;
	public abstract Channel setupChannel()throws IOException;
}
